#include <algorithm>
#include <iostream>
#include <list>
#include <string>
#include <mutex>
#include <assert.h>
#include <vector>
#include <memory>
#include <queue>
#include <stack>
#include <thread>
#include "io_pool.h"

namespace {

    class session : public std::enable_shared_from_this<session>, noncopyable {
    public:
        session(boost::asio::io_context &ios, size_t block_size)
                : io_context_(ios), socket_(ios), block_size_(block_size), buffer_(new char[block_size]) {
        }

        ~session() {
            delete[] buffer_;
        }

        boost::asio::ip::tcp::socket &socket() {
            return socket_;
        }

        void start() {
            boost::asio::ip::tcp::no_delay no_delay(true);
            socket_.set_option(no_delay);
            read();
        }

        void write() {
            boost::asio::async_write(socket_, boost::asio::buffer(buffer_, block_size_),
                                     [this, self = shared_from_this()](
                                             const boost::system::error_code &err, size_t cb) {
                                         if (!err) {
                                             assert(cb == block_size_);
                                             read();
                                         }
                                     });
        }

        void read() {
            boost::asio::async_read(socket_, boost::asio::buffer(buffer_, block_size_),
                                    [this, self = shared_from_this()](
                                            const boost::system::error_code &err, size_t cb) {
                                        if (!err) {
                                            assert(cb == block_size_);
                                            write();
                                        }
                                    });
        }

    private:
        boost::asio::io_context &io_context_;
        boost::asio::ip::tcp::socket socket_;
        size_t const block_size_;
        char *buffer_;
    };

    class server : noncopyable {
    public:
        server(int thread_count, const boost::asio::ip::tcp::endpoint &endpoint,
               size_t block_size)
                : thread_count_(thread_count), block_size_(block_size), context_pool_(thread_count),
                  acceptor_(context_pool_.get_io_context()) {
            acceptor_.open(endpoint.protocol());
            acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(1));
            acceptor_.bind(endpoint);
            acceptor_.listen();
        }

        void start() {
            accept();
        }

        void wait() {
            context_pool_.run();
        }

    private:
        void accept() {
            std::shared_ptr<session> new_session(new session(
                    context_pool_.get_io_context(), block_size_));
            auto &socket = new_session->socket();
            acceptor_.async_accept(socket,
                                   [this, new_session = std::move(new_session)](
                                           const boost::system::error_code &err) {
                                       if (!err) {
                                           std::cerr << "[thread " << std::this_thread::get_id() << "] accept socket"
                                                     << std::endl;
                                           new_session->start();
                                           accept();
                                       }
                                   });
        }

    private:
        int const thread_count_;
        size_t const block_size_;
        io_context_pool context_pool_;
        boost::asio::ip::tcp::acceptor acceptor_;
    };

} // namespace 

void server_test1(int thread_count, char const *host, char const *port,
                  size_t block_size) {
    auto endpoint = boost::asio::ip::tcp::endpoint(
            boost::asio::ip::make_address(host), atoi(port));
    server s(thread_count, endpoint, block_size);
    s.start();
    s.wait();
}


void endless_server() {
    std::vector<std::shared_ptr<std::thread> > threads;
    for (std::size_t i = 0; i < 10; ++i) {
        std::shared_ptr<std::thread> thread(new std::thread(
                []() {
                    std::cout << "Thread " << " is starting.\n";
                    boost::asio::io_context io_context;

                    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard =
                            boost::asio::make_work_guard(io_context);
                    io_context.run();
                    std::cout << "Thread " << " has finished.\n";
                }));
        threads.push_back(thread);
    }

    // Wait for all threads in the pool to exit.
    for (std::size_t i = 0; i < threads.size(); ++i)
        threads[i]->join();
}

int main() {
    server_test1(10, "127.0.0.1", "8888", 64);

}
